#!/usr/bin/env node
// @ts-check
/** @ts-ignore @type {cluster.Cluster} */
const cluster = require('cluster')

/** @typedef {(...params: any[]) => any | Promise<any>} WorkerAction */
/**
 * @typedef {Object} UClusterOptions - Project options
 * @property {number} maxWorkerNum 
 * @property {number} maxTimeout Timeout in ms, 0 to disable
 * @property {{action: string, params: any}[]} jobs
 * @property {{[action: string]: WorkerAction}} worker Job handles
 */
/** @type {UClusterOptions} */
const UClusterDefaultOptions = {
    maxWorkerNum: Math.max(require('os').cpus().length, 2),
    maxTimeout: 0,
    jobs: [],
    worker: {},
}

/**
 * @param {*} obj anything
 * @returns {boolean}
 */
function isFunction(obj) {
    return Object.prototype.toString.call(obj) === '[object Function]'
}

/**
 * @param {*} obj anything
 * @returns {boolean}
 */
function isAsyncFunction(obj) {
    return Object.prototype.toString.call(obj) === '[object AsyncFunction]'
}

/**
 * @namespace UCluster cluster wrap for ufo_parse.js
 */
class UCluster {
    /** @type {number} */
    workerNum

    /**
     * Create a trace parsing project
     * @param {Partial<UClusterOptions>=} config 
     */
    async project(config) {
        /** @type {UClusterOptions} */
        const cfg = { ...UClusterDefaultOptions, ...config }
        if (this.isPrimary) {
            return await this.createPrimary(cfg)
        }
        else if (this.isWorker) {
            this.createWorker(cfg)
        }
    }

    /**
     * @readonly @returns {boolean}
     */
    get isPrimary() {
        const nodeVer = Number(process.versions.node.split('.')[0])
        return (nodeVer >= 16) ? cluster.isPrimary : cluster.isMaster
    }

    /**
     * @readonly @returns {boolean}
     */
    get isWorker() {
        return cluster.isWorker
    }


    /**
     * @param {number} timeout 0 = infinite
     */
    async waitClusterRun(timeout = 0) {
        const maxTime = timeout === 0 ? 0 : (Date.now() + timeout)
        return new Promise(resolve => {
            /** @type {NodeJS.Timer} */
            let timer
            function deamon() {
                if (maxTime && Date.now() > maxTime) {
                    clearInterval(timer)
                    resolve(0)
                }
                if (Object.values(cluster.workers).length === 0) {
                    clearInterval(timer)
                    resolve(0)
                }
            }
            timer = setInterval(deamon, 1000)
        })
    }

    /**
    * Primary process
    * @param {UClusterOptions} options 
    */
    async createPrimary(options) {
        console.log('primary running', process.pid, '#jobs', options.jobs.length)
        // console.log(options);
        try {
            this.workerNum = options.maxWorkerNum
            let jobNum = options.jobs.length
            let jobNumStat = jobNum
            let results = Array(jobNum)
            for (let i = 0; i < this.workerNum; i++) {
                // console.log('fork', i)
                cluster.fork().on('message', (/** @type {{wid: number, jobid: number, result: object}} */ { wid, jobid, result }) => {
                    const worker = cluster?.workers?.[wid]
                    if (!worker) {
                        return
                    }
                    if (jobid >= 0) {
                        results[jobid] = result
                        if (jobNumStat - jobNum > 500) {
                            jobNumStat = jobNum
                            // print progress
                            const progress = (1 - jobNum / options.jobs.length) * 100
                            console.log(`progress ${progress.toFixed(1)}%`)
                        }
                    }
                    // console.log('job', jobid, jobNum)
                    if (jobNum > 0) {
                        const jobid = --jobNum
                        /** @type {object} */
                        let job = options.jobs[jobid]
                        job.jobid = jobid
                        worker.send(job)
                    } else {
                        worker.disconnect()
                    }
                })
            }
            await this.waitClusterRun(options.maxTimeout)
            return results
        } catch (e) {
            console.error(e)
        }
    }

    /**
     * Worker process
     * @param {UClusterOptions} options 
     */
    createWorker(options) {
        for (const action of Object.keys(options.worker)) {
            const func = options.worker[action]
            if (!isFunction(func) && !isAsyncFunction(func)) {
                console.error('should provide a callback function for worker', action)
                return;
            }
        }
        if (process.send && cluster.worker) {
            const wid = cluster.worker.id;
            process.on('message', async (/** @type{{action: string, jobid: number, params: object}} */ { action, jobid, params }) => {
                try {
                    if (options.worker.hasOwnProperty(action) && process.send) {
                        const result = await options.worker[action](...params)
                        process.send({ wid, jobid, result })
                    }
                }
                catch (e) {
                    console.error(e)
                }
            })
            // process.on('disconnect', () => { console.log('worker disconnect', wid) })
            // console.log('worker online', wid, options.worker)
            process.send({ wid, jobid: -1 }) // worker ready
        }
    }
}

module.exports = UCluster